package com.vincent.replica;

import com.vincent.common.*;
import com.vincent.common.Process;
import com.vincent.message.DecisionMessage;
import com.vincent.message.ProposeMessage;
import com.vincent.message.RequestMessage;
import com.vincent.utils.Config;
import com.vincent.utils.MyQueue;
import com.vincent.utils.ReconfigCommand;
import com.vincent.utils.RequestCommand;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by wensen on 2016/12/8.
 */
public class Replica extends Process{

    private static Logger LOGGER = LoggerFactory.getLogger(Replica.class);

    //初始为1，表示第一个slot可用
    private long slot_in = 1;
    //初始为1，表示第一个slot为未
    private long slot_out = 1;
    //proposals, 包含了从request转换到proposal的命令
    private Map<Long, Command> proposals = new HashMap<Long, Command>();
    //包含已经decision的命令
    private Map<Long, Command> decisions = new HashMap<Long, Command>();
    //client的requests
    private MyQueue<Command> requests = new MyQueue<Command>();
    //
    private Config conf;
    public Replica(Environment env, int id, Config conf) {
        super(env, id);
        this.conf = conf;
        this.addProc(this);
    }


    public void propose() {
        /**
         * 该函数将requests转换成proposals
         * 使用slot_in来寻找未使用的slot, 在已知的configuration下
         * slot_out寻找未decision的command,即已经propose但还未decide的command
         * 对于每一个slot, 首先检查该slot的configuration是否不同于前一个slot的configuration
         * 检查方法是：检测decision在(slot_in - WINDOW)这个slot中是否是一个reconfiguration 命令
         * 若是，则为configuration更新slot为s
         * 然后，该函数从requests中取出头部命令，将其放入proposals中，以slot_in为键值
         * 最终，发送这条Propose信息到所有的leaders（在slot_in中configuration）
         */
        while (requests.size() != 0 &&
                this.slot_in < this.slot_out + Constant.WINDOW) {

            /**
             * reconfiguration命令和普通命令类似，但它知道WINDOW个slot之后才生效
             * 这样可允许WINDOW个slot中有待定的proposals
             */
            LOGGER.debug("Replica: propose()");
            if (this.slot_in > Constant.WINDOW &&
                    this.decisions.containsKey(this.slot_in - Constant.WINDOW)) {
               if (this.decisions.get(this.slot_in - Constant.WINDOW) instanceof ReconfigCommand) {

                    ReconfigCommand rcmd = (ReconfigCommand) this.decisions.get(this.slot_in-Constant.WINDOW);
                    List<Integer> r = rcmd.config().replicas();
                    List<Integer> a = rcmd.config().acceptors();
                    List<Integer> l = rcmd.config().leaders();
                    this.conf = new Config();
                    this.conf.acceptors(a);
                    this.conf.replicas(r);
                    this.conf.leaders(l);
                   LOGGER.debug("id: {}, new config: {}", this.id(), this.conf);
                }
            }
            if (!this.decisions.containsKey(this.slot_in)) {
                //将request信息转化为proposal信息，只处理一个slot
                //因为一个slot组成了一个paxos实例(instance)
                LOGGER.debug("Replica: convert request to proposal");

                RequestCommand command = (RequestCommand) this.requests.dequeue();
                this.proposals.put(this.slot_in, command);
                List<Integer> leaders = this.conf.leaders();
                //给每一个leader都发送Propose message
                for (Integer leader : leaders) {
                    this.sendMessage(leader, new ProposeMessage(this.id(), this.slot_in, command));
                }
            }


        }


    }

    /**
     * 该函数和所有replica触发的顺序是一样的，
     * 首先，它会检测该命令前面已经执行过，因为不同的replica可能最终为不同的slot触发相同的命令
     *  这样，相同的命令会决定多次
     *
     * @param cmd
     */
    public void perform(Command cmd) {

        //可能效率不高，但。。。
        for (int s = 1; s < this.slot_out; s++) {
            if (this.decisions.get(s) == cmd) {
                this.slot_out += 1;
                return;
            }
        }

        if (cmd instanceof ReconfigCommand) {
            this.slot_out += 1;
            return;
        }

        this.slot_out += 1;
    }


    /**
     *replica在一个无限的循环中运行，接收信息，repliaca接收两种信息：
     * 1、Requests: 当它从client接收Request时，replica将该request插入requests中，下一步
     *  则会触发propose()函数
     *
     * 2、Decisions:Decision可能乱序或多次到达，对于每一条decision信息，replica将它插入decisions集合中
     * 下一步，在一个循环中，在试图接收更多信息前，考虑哪些decisions已经准备好执行
     * 如果有一个decision刚好对应当前的slot_out，则replica首先会检测是否该replica在这个slot中propose过
     * 不同的命令(即在proposals集合中)，若是，则replica将把该命令移除并插入requests集合中共后面执行
     * 下一步，执行perform()函数
     */
    public void body() {

        LOGGER.debug("Replica: Here I am: {}", this.id());
        while (true) {

            Message msg = null;
            try {
//                TimeUnit.SECONDS.sleep(20);
                msg = this.getNextMessage();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (msg instanceof RequestMessage) {
                this.requests.enqueue(((RequestMessage) msg).command());
                LOGGER.debug("Replica: recvd RequestMessage");
            } else if (msg instanceof DecisionMessage) {

                this.decisions.put(((DecisionMessage) msg).slot_number(), ((DecisionMessage) msg).command());
                while (this.decisions.containsKey(this.slot_out)) {
                    if (this.proposals.containsKey(this.slot_out)) {
                        if (this.proposals.get(this.slot_out) != this.decisions.get(this.slot_out)) {
                            this.requests.enqueue(this.proposals.get(this.slot_out));
                        }
                        this.proposals.remove(this.slot_out);
                    }
                    this.perform(this.decisions.get(this.slot_out));
                }
            }
            else {
                LOGGER.debug("Replica: unknown msg type");
            }
            this.propose();
        }
    }
}